package com.herohead.backup;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.TreeSet;

import org.json.simple.JSONArray;
import org.json.simple.JSONObject;
import org.json.simple.parser.JSONParser;
import org.json.simple.parser.ParseException;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Tuple;

import com.herohead.Misc;
import com.herohead.redis.GetRedisServer;
import com.herohead.redis.RedisKeys;
import com.herohead.redis.RedisVars;

public class CloneStatsV2 extends TimerTask {

	@Override
	public void run() {
		remoteProblem = false;
		//timer.cancel();
	}


	static Jedis jd;
	static Map <Integer, Double> lastValues = new HashMap<Integer, Double>();
	static Map <String, Double> md5Running = new HashMap<String, Double>();
	static Map <String, Double> md5Queued = new HashMap<String, Double>();
	static Timer timer;
	public static boolean remoteProblem = false;
	static TimerTask task;


	public static void main(String[] args) {
		// TODO Auto-generated method stub

		//updateValue("del/craigisgreat");

		//System.setProperty("http.proxyHost", "us00sprxv00.am.mt.mtnet");
		System.setProperty("http.proxyHost", "proxy.am.mt.mtnet");
		System.setProperty("http.proxyPort", "8080");

		Misc m = new Misc();

		GetRedisServer grs = new GetRedisServer();

		//String redisHostName = grs.getRedisHost();
		RedisVars.redisHost = grs.getRedisHost();
		//RedisVars.redisHost = "172.21.5.41";


		jd = new Jedis(RedisVars.redisHost,6380);

		try {
			if (!grs.checkForRedis(jd)) {
				System.out.println("redis server not found");
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}


		//checkServerList("z_prod:clone:queued:size");
		//System.exit(0);


		while (true) {

			if (remoteProblem) {
				System.out.println("----------------------------------------START - remote problem  ----------");
			} else {
				System.out.println("----------------------------------------START------------------------------");
			}

			if (updateRemote(1,String.valueOf(jd.llen(RedisKeys.listProdThreadRunning)),"clone:threads:running")) {
				System.out.print("*");
			}
			System.out.println("Threads running : \t" + jd.llen(RedisKeys.listProdThreadRunning));
			//updateRemote(1,String.valueOf(jd.llen(RedisKeys.listProdThreadRunning)),"clone:threads:running");

			if (updateRemote(2,String.valueOf(jd.hlen(RedisKeys.hashProdThreadQueuedProcessID)),"clone:threads:queued")) {
				System.out.print("*");
			}

			System.out.println("Threads queued : \t" + jd.hlen(RedisKeys.hashProdThreadQueuedProcessID));
			//updateValue("set/" + "clone:threads:queued" + "/" + jd.hlen(RedisKeys.hashProdThreadQueuedProcessID));
			//updateRemote(2,String.valueOf(jd.hlen(RedisKeys.hashProdThreadQueuedProcessID)),"clone:threads:queued");

			if (updateRemote(3,String.valueOf(jd.llen(RedisKeys.listProdThreadCompleted)),"clone:threads:completed")) {
				System.out.print("*");
			}
			System.out.println("Threads completed : \t" + jd.llen(RedisKeys.listProdThreadCompleted));
			//updateValue("set/" + "clone:threads:completed" + "/" + jd.llen(RedisKeys.listProdThreadCompleted));
			//updateRemote(3,String.valueOf(jd.llen(RedisKeys.listProdThreadCompleted)),"clone:threads:completed");

			System.out.println("Threads failed : \t" + jd.llen(RedisKeys.listProdThreadFailed));
			//System.out.println("Clones running : \t" + jd.llen("prod:clones:running"));
			System.out.println("----");

			if (updateRemote(4,jd.llen(RedisKeys.listProdCloneRunning).toString(),"clone:clones:running")) {
				System.out.print("*");
			}
			System.out.println("Clones running : \t" + jd.llen(RedisKeys.listProdCloneRunning));
			//updateValue("set/" + "clone:clones:running" + "/" + jd.llen(RedisKeys.listProdCloneRunning));
			//updateRemote(4,jd.llen(RedisKeys.listProdCloneRunning).toString(),"clone:clones:running");

			if (updateRemote(5,jd.llen(RedisKeys.lProdCloneCompleted).toString(),"clone:clones:completed")) {
				System.out.print("*");
			}
			System.out.println("Clones completed : \t" + jd.llen(RedisKeys.lProdCloneCompleted));
			//updateValue("set/" + "clone:clones:completed" + "/" + jd.llen(RedisKeys.lProdCloneCompleted));
			//updateRemote(5,jd.llen(RedisKeys.lProdCloneCompleted).toString(),"clone:clones:completed");

			if (updateRemote(6,getSize(RedisKeys.hashProdThreadQueuedProcessID),"clone:total:queued:size")) {
				System.out.print("*");
			}
			System.out.println("Total Queued size : \t" + getSize(RedisKeys.hashProdThreadQueuedProcessID));
			//updateValue("set/" + "clone:total:queued:size" + "/" + getSize(RedisKeys.hashProdThreadQueuedProcessID));
			//updateRemote(6,getSize(RedisKeys.hashProdThreadQueuedProcessID),"clone:total:queued:size");

			if (updateRemote(7,getSize(RedisKeys.hashProdThreadRunningProcessID),"clone:total:running:size")) {
				System.out.print("*");
			}
			System.out.println("Total running size : \t" + getSize(RedisKeys.hashProdThreadRunningProcessID));
			//updateValue("set/" + "clone:total:running:size" + "/" + getSize(RedisKeys.hashProdThreadRunningProcessID));
			//updateRemote(7,getSize(RedisKeys.hashProdThreadRunningProcessID),"clone:total:running:size");

			if (updateRemote(8,getSize(RedisKeys.listProdThreadCompletedProcessID),"clone:total:processed:size")) {
				System.out.print("*");
			}
			System.out.println("Total processed size : \t" + getSize(RedisKeys.listProdThreadCompletedProcessID));
			//updateValue("set/" + "clone:total:processed:size" + "/" + getSize(RedisKeys.listProdThreadCompletedProcessID));
			//updateRemote(8,getSize(RedisKeys.listProdThreadCompletedProcessID),"clone:total:processed:size");

			if (updateRemote(9,jd.scard(RedisKeys.setProdSSIDNotCloned).toString(),"clone:ssid:not:cloned")) {
				System.out.print("*");
			}
			System.out.println("SSID not cloned : \t" + jd.scard(RedisKeys.setProdSSIDNotCloned));
			//updateValue("set/" + "clone:ssid:not:cloned" + "/" + jd.scard(RedisKeys.setProdSSIDNotCloned));
			//updateRemote(9,jd.scard(RedisKeys.setProdSSIDNotCloned).toString(),"clone:ssid:not:cloned");

			if (updateRemote(10,jd.scard(RedisKeys.setProdGroupNotCloned).toString(),"clone:groups:not:cloned")) {
				System.out.print("*");
			}
			System.out.println("Groups not cloned : \t" + jd.scard(RedisKeys.setProdGroupNotCloned));
			//updateValue("set/" + "clone:groups:not:cloned" + "/" + jd.scard(RedisKeys.setProdGroupNotCloned));
			//updateRemote(10,jd.scard(RedisKeys.setProdGroupNotCloned).toString(),"clone:groups:not:cloned");

			if (updateRemote(15,jd.get(RedisKeys.strProdSizeNotCloned),"clone:size:not:cloned")) {
				System.out.print("*");
			}
			System.out.println("Size not cloned : \t" + m.bytesToMB(jd.get(RedisKeys.strProdSizeNotCloned)) + " MB");
			//System.out.println("Size not cloned : \t" + m.bytesToMB(jd.get(RedisKeys.strProdSizeNotCloned)) + " MB");




			if (updateRemote(11,m.bytesToMB(getSmallestInHash(RedisKeys.hashProdThreadRunningProcessID)).toString(),"clone:smallest:running")) {
				System.out.print("*");
			}
			System.out.println("Smallest running : \t" + m.bytesToMB(getSmallestInHash(RedisKeys.hashProdThreadRunningProcessID)) + " MB");
			//updateValue("set/" + "clone:smallest:running" + "/" + m.bytesToMB(getSmallestInHash(RedisKeys.hashProdThreadRunningProcessID)));
			//updateRemote(11,m.bytesToMB(getSmallestInHash(RedisKeys.hashProdThreadRunningProcessID)).toString(),"clone:smallest:running");

			if (updateRemote(12,m.bytesToMB(getLargestInHash(RedisKeys.hashProdThreadRunningProcessID)).toString(),"clone:largest:running")) {
				System.out.print("*");
			}
			System.out.println("Largest running : \t" + m.bytesToMB(getLargestInHash(RedisKeys.hashProdThreadRunningProcessID)) + " MB");
			//updateValue("set/" + "clone:largest:running" + "/" + m.bytesToMB(getLargestInHash(RedisKeys.hashProdThreadRunningProcessID)));
			//updateRemote(12,m.bytesToMB(getLargestInHash(RedisKeys.hashProdThreadRunningProcessID)).toString(),"clone:largest:running");

			//System.out.println("Smallest running : \t" + (getSmallestInHash(RedisKeys.hashProdThreadRunningProcessID)) + " MB");
			//System.out.println("Largest running : \t" + (getLargestInHash(RedisKeys.hashProdThreadRunningProcessID)) + " MB");
			//jd.del("temp:hash:list");
			//createTempTable(RedisKeys.hashProdThreadQueuedProcessID);
			if (updateRemote(13,m.bytesToMB(getSmallestInHash(RedisKeys.hashProdThreadQueuedProcessID)).toString(),"clone:smallest:in:queue")) {
				System.out.print("*");
			}
			System.out.println("Smallest in Queue : \t" + m.bytesToMB(getSmallestInHash(RedisKeys.hashProdThreadQueuedProcessID)) + " MB");
			//updateValue("set/" + "clone:smallest:in:queue" + "/" + m.bytesToMB(getSmallestInHash(RedisKeys.hashProdThreadQueuedProcessID)));
			//updateRemote(13,m.bytesToMB(getSmallestInHash(RedisKeys.hashProdThreadQueuedProcessID)).toString(),"clone:smallest:in:queue");

			if (updateRemote(14,m.bytesToMB(getLargestInHash(RedisKeys.hashProdThreadQueuedProcessID)).toString(),"clone:largest:in:queue")) {
				System.out.print("*");
			}
			System.out.println("Largest in Queue : \t" + m.bytesToMB(getLargestInHash(RedisKeys.hashProdThreadQueuedProcessID)) + " MB");
			//updateValue("set/" + "clone:largest:in:queue" + "/" + m.bytesToMB(getLargestInHash(RedisKeys.hashProdThreadQueuedProcessID)));
			//updateRemote(14,m.bytesToMB(getLargestInHash(RedisKeys.hashProdThreadQueuedProcessID)).toString(),"clone:largest:in:queue");

			//System.out.println("Smallest in Queue : \t" + (getSmallestInHash(RedisKeys.hashProdThreadQueuedProcessID)) + " MB");
			//System.out.println("Largest in Queue : \t" + (getLargestInHash(RedisKeys.hashProdThreadQueuedProcessID)) + " MB");
			//jd.del("temp:hash:list");
			//System.out.println("Clones Names : \n" + getServer("prod:clones:running:name"));
			//System.out.println("\n-----Clones Names");
			System.out.println(getServer(RedisKeys.hashProdCloneRunningName));

			/**
			String str = IOUtils.toString(s, "UTF-8");
			String str = null;
			try {
				str = new String(s);
				System.out.println(str);
			} catch (UnsupportedEncodingException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
			//updateValue("restore/" + RedisKeys.hashProdCloneRunningName + "/0/" + str);
			//System.out.println("-----Queue Sizes");
			 **/


			System.out.println("-----Running");

			checkServerListV2(md5Running,"z_prod:clone:running:size");
			System.out.println("-----Queued");
			checkServerListV2(md5Queued,"z_prod:clone:queued:size");

			System.out.println("-----------------------------------");

			if (!remoteProblem) cleanUpRemote();

			jd.close();

			try {
				Thread.sleep(10000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}


	}

	@SuppressWarnings("null")
	public static String getSmallestInHash(String hash) {



		TreeSet<Long> set = new TreeSet<Long>();

		double d = 0;
		List<String>  s = jd.hvals(hash);
		for (String ss : s) {
			set.add(Long.parseLong(ss));
			//System.out.println(ss);
		}
		if (set.size() == 0) {
			return "0";
		} else {
			return set.first().toString();
		}
		//System.out.println(set.first());
		//String sss = String.valueOf(d);

	}


	public static String getLargestInHash(String hash) {

		TreeSet<Long> set = new TreeSet<Long>();

		double d = 0;
		List<String>  s = jd.hvals(hash);
		for (String ss : s) {
			set.add(Long.parseLong(ss));
			//System.out.println(ss);
		}
		if (set.size() == 0) {
			return "0";
		} else {
			return set.last().toString();
		}
		//System.out.println(set.last());
		//String sss = String.valueOf(d);

	}


	public static void getQueSizing() {

		GetRedisServer grs = new GetRedisServer();

		RedisVars.redisHost = grs.getRedisHost();
		//RedisVars.redisHost = "172.21.5.41";
		RedisVars.redisHost = grs.getHostName();
		Jedis jdro = new Jedis(RedisVars.redisHost,6380);
		Jedis jd = new Jedis(RedisVars.redisHost);
		//System.out.println("Started");
		Set<String> s;
		try {
			s = jdro.hkeys(RedisKeys.hashProdThreadQueuedProcessID);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			return;
		}



		//List<String> runningClones = jdro.hvals(RedisKeys.hashProdCloneRunningName);
		jd.del("z_temp:inprocess:queued:list");
		jd.del("z_temp:inprocess:running:list");
		jd.del("z_temp:inprocess:totalsize");

		for (String processID : s) {

			//System.out.println(processID);
			String command = (jdro.hget(RedisKeys.hashProdThreadQueuedCommand, processID));
			command = command.replace("|", ":");
			//System.out.println(command);
			String cloneName = command.split(":")[0];

			//System.out.println(jdro.hget(RedisKeys.hashProdThreadQueuedProcessID,processID));
			jd.zincrby("z_temp:inprocess:queued:list",1, cloneName);
			try {
				jd.zincrby("z_temp:inprocess:totalsize",Double.parseDouble(jdro.hget(RedisKeys.hashProdThreadQueuedProcessID,processID)), cloneName);
			} catch (Exception e) {
				//catch (NumberFormatException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

		}

		Set<Tuple> tup = jd.zrangeWithScores("z_temp:inprocess:queued:list", 0, -1);
		//Set<Tuple> tupTotal = jd.zrangeWithScores("z_temp:inprocess:totalsize", 0, -1);
		System.out.println("-----Queue Sizes (" + tup.size() + ")");
		for (Tuple s1 : tup) {
			System.out.print(s1.getElement() + "\t");
			System.out.print(String.format("%.0f",s1.getScore()) + "\t");
			System.out.println(String.format("%.0f",jd.zscore("z_temp:inprocess:totalsize", s1.getElement())/1024/1024));
			//System.out.println(tupTotal     .s1.getElement());
		}

		jdro.close();
		jd.close();
	}


	public static void updateValue (String command) {


		if (!remoteProblem) {

			try {




				String url = "http://herohead.tzo.net:7379/" + command;
				URL obj = new URL(url);
				HttpURLConnection con = (HttpURLConnection) obj.openConnection();
				//con.setConnectTimeout(1);
				//System.out.println(con.getConnectTimeout());


				con.setReadTimeout(5000);
				con.setRequestMethod("PUT");
				con.getContent();
				BufferedReader in = new BufferedReader(
						new InputStreamReader(con.getInputStream()));
				String inputLine;
				StringBuffer response = new StringBuffer();

				while ((inputLine = in.readLine()) != null) {
					response.append(inputLine);
				}
				in.close();
				//System.out.println(response.toString());
			} catch (Exception e) {
				remoteProblem = true;
				task = new CloneStatsV2();
				timer = new Timer();
				timer.schedule(task, 100000);

				//e.printStackTrace();
			}

		}
	}


	public static boolean updateRemote(int indexNumber ,String value,String key) {

		double lastV;
		boolean updated = false;

		double passV = Double.parseDouble(value);

		if (lastValues.containsKey(indexNumber)) {
			lastV = lastValues.get(indexNumber);
			lastValues.put(indexNumber, passV);
			if (lastV != passV) {
				//System.out.println("Something changed " + key);
				if (!remoteProblem) updateValue("set/" + key + "/" + value);
				updated=true;;
			}
		} else {
			lastValues.put(indexNumber, passV);
			//System.out.println("Something changed " + key);
			if (!remoteProblem) updateValue("set/" + key + "/" + value);
			updated = true;
		}

		return updated;

	}


	public static boolean checkServerList(Map <String, Double>passedMap, String key) {

		Double lastV;
		//System.out.println("In ServerList");
		Set<Tuple> runningSize = jd.zrevrangeWithScores(key, 0, -1);
		Boolean updated = false;
		//updateValue("del/" + key);
		for (Tuple d : runningSize) {
			String el = d.getElement();
			double score = d.getScore();
			//System.out.println(el + ":" + score);
			if (passedMap.containsKey(el)) {
				lastV = passedMap.get(el);

				//System.out.println(lastV);
				//System.out.println(lastV);
				if (lastV != score) {
					//System.out.println("Something changed " + key);
					passedMap.put(el, score);
					if (score == 0.0) {
						if (!remoteProblem) updateValue("zrem/" + key + "/" + el);
					} else {
						if (!remoteProblem) updateValue("zadd/" + key + "/" + score + "/" + el);
					}
					updated=true;

				}
			} else {
				passedMap.put(el, score);
				if (score == 0.0) {
					if (!remoteProblem) updateValue("zrem/" + key + "/" + el);
				} else {
					if (!remoteProblem) updateValue("zadd/" + key + "/" + score + "/" + el);
				}

				//System.out.println("Something changed " + key);
				//updateValue("zadd/" + key + "/" + score + "/" + el);
				updated = true;
			}


		}

		return updated;

	}




	public static String getServer (String hash) {

		String temp = "";
		Map<String, String> mp = jd.hgetAll(hash);
		System.out.println("\n-----Clones Names (" + mp.size() + ")");
		Iterator it = mp.entrySet().iterator();
		while (it.hasNext()) {
			Map.Entry pairs = (Map.Entry)it.next();
			//System.out.println(pairs.getKey() + " = " + pairs.getValue());
			temp = temp + pairs.getValue().toString() + "\n";
			//size = size + Long.parseLong(temp);
		}

		//Misc m = new Misc();

		//String s = m.bytesToMB(String.valueOf(size)) + " MB";

		return temp;
	}


	@SuppressWarnings("rawtypes")
	public static String getSize (String hash) {

		long size = 0;

		Map<String, String> mp = jd.hgetAll(hash);

		Iterator it = mp.entrySet().iterator();
		while (it.hasNext()) {
			Map.Entry pairs = (Map.Entry)it.next();
			//System.out.println(pairs.getKey() + " = " + pairs.getValue());
			String temp = pairs.getValue().toString();
			size = size + Long.parseLong(temp);
		}

		Misc m = new Misc();

		String s = m.bytesToMB(String.valueOf(size));

		return s;
	}

	public static boolean checkServerListV2(Map <String, Double>passedMap, String key) {

		Double lastV;
		//System.out.println("In ServerList");
		Set<Tuple> runningSize = jd.zrevrangeWithScores(key, 0, -1);
		Boolean updated = false;

		Misc m = new Misc();

		for (Tuple d : runningSize) {
			String el = d.getElement();
			double score = d.getScore();
			//System.out.println(el + ":" + score);
			if (passedMap.containsKey(el)) {
				lastV = passedMap.get(el);

				//System.out.println(lastV);
				//System.out.println(lastV);
				if (lastV != score) {
					if (String.format("%.0f",score).contentEquals("0")) {
						if (!remoteProblem) updateValue("zrem/" + key + "/" + el);
					} else {
						if (!remoteProblem) updateValue("zadd/" + key + "/" + score + "/" + el);
						//System.out.println("Something changed " + key);
						System.out.println(formatTabSpacing("*" + el,String.format("%.0f", score/1024/1204)));
						//System.out.println("*" + el + "\t" + String.format("%.0f", score/1024/1204));
					}

					passedMap.put(el, score);
					updated=true;
				} else {
					if (!String.format("%.0f",score).contentEquals("0")) {
						//System.out.println(el + "\t" + String.format("%.0f", score/1024/1204));
						System.out.println(formatTabSpacing(el,String.format("%.0f", score/1024/1204)));
					}
				}
			} else {
				passedMap.put(el, score);
				if (String.format("%.0f",score).contentEquals("0")) {
					if (!remoteProblem) updateValue("zrem/" + key + "/" + el);
				} else {
					if (!remoteProblem) updateValue("zadd/" + key + "/" + score + "/" + el);
					//System.out.println("Something changed " + key);
					//System.out.println("*" + el + "\t" + String.format("%.0f", score/1024/1204));
					System.out.println(formatTabSpacing("*" + el,String.format("%.0f", score/1024/1204)));
				}

				updated = true;
			}


		}

		return updated;

	}
	public static String getValue (String command) {

		String ret = null;

		if (!remoteProblem) {

			try {

				String url = "http://herohead.tzo.net:7379/" + command;
				//String url = "http://65.60.174.116:7379/" + command;

				URL obj = new URL(url);

				HttpURLConnection con = (HttpURLConnection) obj.openConnection();

				//System.setProperty("http.proxyHost", "us00sprxv00.am.mt.mtnet");
				//System.setProperty("http.proxyPort", "8080");
				con.setRequestMethod("PUT");

				BufferedReader in = new BufferedReader(
						new InputStreamReader(con.getInputStream()));
				String inputLine;
				StringBuffer response = new StringBuffer();

				while ((inputLine = in.readLine()) != null) {
					response.append(inputLine);
				}
				in.close();
				ret = response.toString();
			} catch (Exception e) {
				e.printStackTrace();

			}
		}
		//System.out.println(ret);
		return ret;
	}


	public static void cleanUpRemote() {

		if (remoteProblem) {

			JSONParser parser = new JSONParser();
			Object obj = null;
			try {
				obj = parser.parse(getValue("zrevrange/z_prod:clone:running:size/0/-1"));
			} catch (ParseException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}


			JSONObject jsonObject = (JSONObject) obj;

			JSONArray remote = (JSONArray) jsonObject.get("zrevrange");

			Set<String> s = jd.zrevrange("z_prod:clone:running:size", 0, -1);

			Iterator<String> iterator = remote.iterator();

			while (iterator.hasNext()) {
				String temp = iterator.next();
				if (s.contains(temp))	{
					//			System.out.println("Found match " + temp);
				} else {
					getValue("zrem/z_prod:clone:running:size/" + temp);
					//			System.out.println("Not Found " + temp);
				}
			}
		}
	}

	public static String formatTabSpacing(String f, String l) {

		if (f.length() < 6) {
			return f + "\t\t\t" + l;
		} else if (f.length() < 16) {
			return f + "\t\t" + l;
		} else {
			return f + "\t" + l;
		}

	}

}


